package com.xlhj.boot.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author: liucaijing
 * @description: HBaseUtils
 * @date: 2021/6/9 8:55
 */
@Component
public class HBaseUtils {

    private static String quorums;

    @Value("${hbase.config.zookeeper.quorum}")
    private String quorum;

    @PostConstruct
    public void getQuorum() {
        quorums = this.quorum;
    }

    /**
     * 获取hbase连接
     * @return
     * @throws Exception
     */
    public static Connection getHBaseConnection() {
        Connection connection = null;
        try {
            //hbase配置对象
            Configuration configuration = HBaseConfiguration.create();
            //设置zookeeper地址
            configuration.set("hbase.zookeeper.quorum", quorums);
            //获取hbase连接对象
            connection = ConnectionFactory.createConnection(configuration);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return connection;
    }

    /**
     * 获取Admin对象 管理对象
     * @return
     * @throws Exception
     */
    public static Admin getAdmin() throws Exception {
        return getHBaseConnection().getAdmin();
    }

    /**
     * 根据表名获取表对象
     * @param tbName
     * @return
     * @throws Exception
     */
    public static Table getTable(String tbName) throws Exception {
        TableName tableName = TableName.valueOf(tbName);
        Table table = getHBaseConnection().getTable(tableName);
        return table;
    }

    /**
     * 创建表
     * @param tbName
     * @throws Exception
     */
    public static void createTable(String tbName) throws Exception {
        Admin admin = getAdmin();
        TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tbName));
        //列族构建
        ColumnFamilyDescriptorBuilder familyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder("user_info".getBytes());
        ColumnFamilyDescriptor familyDescriptor = familyDescriptorBuilder.build();
        //添加到表中
        builder.setColumnFamily(familyDescriptor);
        TableDescriptor tableDescriptor = builder.build();
        admin.createTable(tableDescriptor);
        //释放资源
        admin.close();
    }

    public static void getTableData(String tbName) throws Exception {
        Table table = getTable(tbName);
        //获取1行数据
        Get get = new Get("row_key".getBytes());
        Result result = table.get(get);
        while (result.advance()) {
            //获取当前单元格
            Cell cell = result.current();
            byte[] row = CellUtil.cloneRow(cell);
            byte[] family = CellUtil.cloneFamily(cell);
            byte[] qualifier = CellUtil.cloneQualifier(cell);
            byte[] value = CellUtil.cloneValue(cell);

        }
        table.close();
    }
}
